package kikoqiu.flex;

import flex.messaging.io.amf.client.AMFConnection;
import flex.messaging.io.amf.client.exceptions.ClientStatusException;
import flex.messaging.io.amf.client.exceptions.ServerStatusException;
import flex.messaging.messages.CommandMessage;
import flex.messaging.util.UUIDUtils;
import flex.messaging.messages.*;

public class Consumer extends ConsumerCallBack implements Runnable{
	private String subTopic;
	private String destination;
	private String endPoint;
	private String url;
	private int pollingInterval=1000;

	private double messageVersion=1.0;
	private AMFConnection amfConnection;
	private ConsumerCallBack callBack=this;
	
	private String DSId;
	private Thread pollingThread;
	

	/**
	 * Constructor
	 * @param url MessageBroker URL
	 * @param destination The Destination
	 * @param subTopic The SubTopic
	 */
	public Consumer(String url,String destination,String subTopic){
		this.subTopic=subTopic;
		this.destination=destination;
		this.endPoint="";//not needed actually
		this.url=url;
		amfConnection=new AMFConnection();
		amfConnection.addHttpRequestHeader("Content-type", "application/x-amf");
	}
	
	/**
	 * Constructor
	 * @param url MessageBroker URL
	 * @param destination The Destination
	 * @param subTopic The SubTopic
	 * @param endPoint The EndPoint(Not needed actually)
	 */
	public Consumer(String url,String destination,String subTopic,String endPoint){
		this.subTopic=subTopic;
		this.destination=destination;
		this.endPoint=endPoint;
		this.url=url;
		amfConnection=new AMFConnection();
		amfConnection.addHttpRequestHeader("Content-type", "application/x-amf");
	}
	
    static{
    	/*
    	 * do class Alias reg
    	 */
    	AMFConnection.registerAlias(flex.messaging.messages.AcknowledgeMessageExt.CLASS_ALIAS, flex.messaging.messages.AcknowledgeMessageExt.class.getName());
    	AMFConnection.registerAlias(flex.messaging.messages.AsyncMessageExt.CLASS_ALIAS, flex.messaging.messages.AsyncMessageExt.class.getName());
    }
    /**
     * Begin subscribe by starting the polling thread
     * @throws ClientStatusException
     * @throws ServerStatusException
     */
	public void subscribe() throws ClientStatusException,ServerStatusException{
		amfConnection.connect(getUrl());
		
		AcknowledgeMessageExt m=(AcknowledgeMessageExt)call(buildPrepare());
		DSId=(String)m.getHeader(CommandMessage.FLEX_CLIENT_ID_HEADER);		
		getCallBack().onAcknowledgeMessageExt(m);
		
		m=(AcknowledgeMessageExt)call(buildSubscribe());		
		getCallBack().onAcknowledgeMessageExt(m);		
		
		pollingThread=new Thread(this);
		pollingThread.start();
	}
	/**
	 * Stop polling thread
	 */
	public void unscribe(){
		Thread oldrun=pollingThread;
		pollingThread=null;
		try {
			oldrun.join(3000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		if(oldrun.getState()!=Thread.State.TERMINATED){
			oldrun.interrupt();
		}
		amfConnection=null;
	}	
	/**
	 * 
	 * @return The Message Version (DSMessageVersion header)
	 */
	public double getMessageVersion() {
		return messageVersion;
	}
	/**
	 * 
	 * @return The AmfConnection
	 */
	public AMFConnection getAmfConnection() {
		return amfConnection;
	}
	/**
	 * 
	 * @return The subscribe Destination
	 */
	public String getDestination() {
		return destination;
	}
	/**
	 * 
	 * @return The SubTopic
	 */
	public String getSubTopic() {
		return subTopic;
	}
	/**
	 * 
	 * @return The EndPoint
	 */
	public String getEndPoint() {
		return endPoint;
	}
	public String getUrl() {
		return url;
	}
	/**
	 * 
	 * @return The Call back class(default is this)
	 * @see setCallBack
	 */
	public ConsumerCallBack getCallBack() {
		return callBack;
	}
	/**
	 * Set the call back class(default is this)
	 * @param callBack
	 * @see getCallBack
	 */
	public void setCallBack(ConsumerCallBack callBack) {
		this.callBack = callBack;
	}
	/**
	 * 
	 * @return The thread that is polling messages
	 */
	public Thread getPollingThread() {
		return pollingThread;
	}
	/**
	 * 
	 * @return the interval between each poll(default is 1000)
	 */
	public int getPollingInterval() {
		return pollingInterval;
	}
	/**
	 * Set the interval between each poll(default is 1000)
	 * @param pollingInterval
	 */
	public void setPollingInterval(int pollingInterval) {
		this.pollingInterval = pollingInterval;
	}
	
	
	
	
	public void run() {
		while(pollingThread!=null){
			try{
				Object o=call(buildPoll());
				if(o instanceof AcknowledgeMessageExt){
					AcknowledgeMessageExt cm=(AcknowledgeMessageExt)o;
					getCallBack().onAcknowledgeMessageExt(cm);
					Thread.sleep(getPollingInterval());
				}else if(o instanceof CommandMessage){
					processCommandMessage(o);
				}else{
					throw new ClientStatusException("Unknown Call Return Type "+o,ClientStatusException.AMF_CALL_FAILED_CODE);
				}				
			}catch(Exception e){	
				getCallBack().onFault(e);
			}
		}		
	}	
	@SuppressWarnings("unchecked")
	private void processCommandMessage(Object o) throws ClientStatusException {
		CommandMessage cm=(CommandMessage)o;
		Object msg=null;
		if(cm.getBody() instanceof Object[]){
			msg=((Object[])cm.getBody())[0];
		}else if(cm.getBody() instanceof java.util.ArrayList){
			java.util.ArrayList<Object> al=(java.util.ArrayList<Object>)cm.getBody();
			msg=al.get(0);
		}else{
			throw new ClientStatusException("Unknown CommandMessage body",ClientStatusException.AMF_CALL_FAILED_CODE);
		}
		getCallBack().onMessage(msg);
	}	
	private Object call(Object o) throws ClientStatusException, ServerStatusException{
		return amfConnection.call("null", o);
	}	
	private CommandMessage buildPrepare() {
		return build(CommandMessage.CLIENT_PING_OPERATION,"nil");
	}
	private CommandMessage buildSubscribe(){
		return build(CommandMessage.SUBSCRIBE_OPERATION,DSId);
	}
	private CommandMessage build(int oId,String DSId) {
		CommandMessage cm=new CommandMessage();
		cm.setOperation(oId);
		cm.setMessageId(UUIDUtils.createUUID());
		cm.setDestination(getDestination());
		cm.setTimestamp(System.currentTimeMillis());
		
		cm.setHeader(CommandMessage.FLEX_CLIENT_ID_HEADER,DSId);
		cm.setHeader(CommandMessage.SUBTOPIC_HEADER_NAME,getSubTopic());
		cm.setHeader(CommandMessage.MESSAGING_VERSION, getMessageVersion());
		cm.setHeader(CommandMessage.ENDPOINT_HEADER, getEndPoint());
		return cm;
	}
	private CommandMessageExt buildPoll() {
		CommandMessageExt cm=new CommandMessageExt();
		cm.setOperation(CommandMessage.POLL_OPERATION);
		cm.setMessageId(UUIDUtils.createUUID());
		cm.setTimestamp(System.currentTimeMillis());
		
		cm.setHeader(CommandMessage.FLEX_CLIENT_ID_HEADER,DSId);
		return cm;
	}
}
